package com.chj.config;

import com.alibaba.fastjson.JSONObject;
import com.chj.constant.Rabbit;
import com.chj.utils.UUIDUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Administrator on 2019/3/5.
 */
@Configuration
public class RabbitConfig {
    private static final Logger logger = LoggerFactory.getLogger(RabbitConfig.class);

//    @Bean
//    public Queue Queue() {
//        return new Queue("hello");
//    }

    public static final Map<String, SimpleMessageListenerContainer> containerMap = new HashMap<String, SimpleMessageListenerContainer>();

    private static ConnectionFactory factory = null;

    @Value("${spring.rabbitmq.host}")
    protected String host;
    @Value("${spring.rabbitmq.port}")
    protected String port;
    @Value("${spring.rabbitmq.username}")
    protected String username;
    @Value("${spring.rabbitmq.password}")
    protected String password;
    @Value("${spring.rabbitmq.virtualHost}")
    protected String virtualHost;

    @Bean(name = "rabbitMqConnectionFactory")
    public ConnectionFactory connectionFactory(){
        // 使用手工方式生成 connection factory，原因是为了设置 publisherConfirms = true
        logger.info("RabbitMQ host {}, port {}, username {}, virtualHost {} ",
                host, port, username, virtualHost);
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setAddresses(host+":"+port);
        cachingConnectionFactory.setUsername(username);
        cachingConnectionFactory.setPassword(password);
        cachingConnectionFactory.setVirtualHost(virtualHost);
        cachingConnectionFactory.setPublisherConfirms(true);
        cachingConnectionFactory.setPublisherReturns(true);
        factory = cachingConnectionFactory;
        return  cachingConnectionFactory;

    }

    @Bean
    public RabbitAdmin rabbitAdmin(){
        return  new RabbitAdmin(connectionFactory());
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(){
        // 必须是prototype类型
        // 如果需要在生产者中添加消息消费后的回调，需要对rabbitTemplate设置ConfirmCallback对象，
        // 由于不同的生产者需要对应不同的ConfirmCallback，如果rabbitTemplate设置为单例bean，
        // 则所有的rabbitTemplate 实际的ConfirmCallback为最后一次申明的ConfirmCallback。
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMandatory(true); //
        return template;
    }

    public static SimpleMessageListenerContainer createListenerContainer(Queue queue, MessageListener listener){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setQueues(queue);
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1); //
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式为手工确认
//        container.setMessageListener(listener); // object listener is Deprecated should use MessageListener replace
        container.setMessageListener(listener);
        return container;
    }


    @Bean(name = Rabbit.Bill.EX_BILL)
    public TopicExchange exchangeBill() {
        return new TopicExchange(Rabbit.Bill.EX_BILL);
    }

    @Bean(name = Rabbit.Contract.EX_CONTRACT)
    public TopicExchange exchangeContract() {
        return new TopicExchange(Rabbit.Contract.EX_CONTRACT);
    }

    @Bean(name = Rabbit.Dict.EX_DICT)
    public TopicExchange exchangeDict() {
        return new TopicExchange(Rabbit.Dict.EX_DICT);
    }

    @Bean(name = Rabbit.Dts.EX_DTS)
    public TopicExchange exchangeDts() {
        return new TopicExchange(Rabbit.Dts.EX_DTS);
    }

    @Bean(name = Rabbit.Funds.EX_FUNDS)
    public TopicExchange exchangeFunds() {
        return new TopicExchange(Rabbit.Funds.EX_FUNDS);
    }

    @Bean(name = Rabbit.Hzcms.EX_HZCMS)
    public TopicExchange exchangeHzcms() {
        return new TopicExchange(Rabbit.Hzcms.EX_HZCMS);
    }

    @Bean(name = Rabbit.Loan.EX_LOAN)
    public TopicExchange exchangeLoan() {
        return new TopicExchange(Rabbit.Loan.EX_LOAN);
    }

    @Bean(name = Rabbit.Message.EX_MESSAGE)
    public TopicExchange exchangeMessage() {
        return new TopicExchange(Rabbit.Message.EX_MESSAGE);
    }

    @Bean(name = Rabbit.Order.EX_ORDER)
    public TopicExchange exchangeOrder() {
        return new TopicExchange(Rabbit.Order.EX_ORDER);
    }

    public static String messageBodyStr(Message message){

        try {
            String encoding = message.getMessageProperties().getContentEncoding();
            if (StringUtils.isBlank(encoding)) {
                encoding = "UTF-8";
            }
            String body = new String(message.getBody(), encoding);
            return body;
        } catch (Exception e) {
            logger.error("{} {}", message, e);
            throw new RuntimeException(e);
        }

    }

    /**
     * RabbitMQ 中的生产者指定的CorrelationData(消费ID)
     * @return
     */
    public static CorrelationData correlationData() {
        String uuid = UUIDUtils.getUUID();
        CorrelationData correlationData = new CorrelationData(uuid);
        return correlationData;
    }

    public static JSONObject messageBody(Message message){
        String body = null;
        try{
            // 转换消息，我们是使用json数据格式
            String encoding = message.getMessageProperties().getContentEncoding();
            if (StringUtils.isBlank(encoding)) {
                encoding = "UTF-8";
            }
            body = new String(message.getBody(), encoding);
            return JSONObject.parseObject(body, JSONObject.class);
        } catch(Exception e){
            logger.error("{} {}", body, e);
        }
        return new JSONObject();

    }

}
